package com.wuwangfu.source;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;

import java.util.Arrays;
import java.util.List;

/**
 * @Author jcshen
 * @Date 2023-02-21
 * @PackageName:com.wuwangfu.source
 * @ClassName: CustomSourceB
 * @Description: 自定义Source
 * @Version 1.0
 */
public class CustomSourceB {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(1);

        DataStreamSource<String> line = env.addSource(new MyBSource());
        System.out.println(line.getParallelism());

        line.print();
        env.execute();

    }

    /**
     * 实现ParallelSourceFunction接口的Source，可以是多并行度
     */
    private static class MyBSource implements ParallelSourceFunction<String> {
        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            List<String> list = Arrays.asList("aaa", "bbb", "ccc", "ddd");
            for (String word : list){
                ctx.collect(word);
            }
        }

        @Override
        public void cancel() {

        }
    }
}
